//
// See the file COPYING for redistribution information.
//

#include <iostream>
#include <sstream>
#include <cstdio>
#include <cstring>
#include <pthread.h>
#include <time.h>
#include <signal.h>
#include <errno.h>
#include <Naipax.hpp>

using namespace std;
using namespace Naipax;

class UrlDispatcher {
public:
    UrlDispatcher() {
        _mpHostHistory = new History("_host_hist");
        _mpSubHostHistory = new History("_host_hist.sub");
    }
    ~UrlDispatcher() {
        delete _mpSubHostHistory;
        delete _mpHostHistory;
    }
    uint32_t operator()(const std::string &rData, uint32_t PriorityLevels) {
        Url UrlObj(rData);
        string SubHostname = UrlObj.SubHostname(2);
        uint32_t PathDepth = UrlObj.PathDepth();

        if (_mpHostHistory->Has(UrlObj.Host()) == true
            || _mpSubHostHistory->Has(SubHostname) == true) {
            return PriorityLevels - 1;
        }
        else if (UrlObj.Query().length() > 0) {
            return PriorityLevels;
        }
        else if (PathDepth > 5) {
            return PriorityLevels;
        }
        else {
            _mpHostHistory->Add(UrlObj.Host(), 60);
            _mpSubHostHistory->Add(SubHostname, 60);
            return PathDepth < PriorityLevels ? PathDepth : PriorityLevels - 1;
        }
    }

private:
    History *_mpHostHistory;
    History *_mpSubHostHistory;
};

struct CallbackData {
    pthread_mutex_t mLock;
    Config *mpConf;
    Connection *mpConn;
    ServiceServer *mpServer;
    PpqScheduler<UrlDispatcher> *mpUrlScheduler;
    History *mpUrlHistory;
};

static CallbackData gsCbData;
static time_t gsStartTime = (time_t) -1;

void ResponseHandler(Connection &rConn,
                     string &rUrl, string &rResponse) {
    static uint64_t sTotalLength = 0;
    static uint32_t sMaxPageRate = 0;
    static uint32_t sMaxDataRate = 0;

    uint32_t PageRate = 0;
    uint32_t DataRate = 0;

    time_t ElapsedTime = time(NULL) - gsStartTime;
    if (ElapsedTime == 0) {
        ElapsedTime = 1;
    }
    sTotalLength += rResponse.length();

    PageRate = rConn.CompletedCount() / ElapsedTime;
    if (PageRate > sMaxPageRate) {
        sMaxPageRate = PageRate;
    }
    DataRate = sTotalLength / ElapsedTime;
    if (DataRate > sMaxDataRate) {
        sMaxDataRate = DataRate;
    }

    cout << "\n"
         << "[Current Connections] " << rConn.CurConnCount() << "\n"
         << "[Page Count] " << rConn.CompletedCount() << "\n"
         << "[Url] " << rUrl << "\n"
         << "[Length] " << rResponse.length() << "\n"
         << "[Elapsed Time] " << ElapsedTime << "\n"
         << "[Page Rate] " << PageRate << "\n"
         << "[Max Page Rate] " << sMaxPageRate << "\n"
         << "[Data Rate] " << DataRate << "\n"
         << "[Max Data Rate] " << sMaxDataRate << "\n"
         << "[Est. Pages/Day] " << PageRate * 86400 << "\n"
         << "\n\n";

    if (gsCbData.mpConf->mCrawlExternalLinks != true) {
        return;
    }

    vector<string> Links;
    vector<string> CheckedLinks;

    LinkExtor LE;
    LE.Extract(rResponse, Links);

    for (vector<string>::iterator LinksIter = Links.begin();
         LinksIter != Links.end(); ++LinksIter) {
        if (gsCbData.mpUrlHistory->Has(*LinksIter) != true) {
            CheckedLinks.push_back(*LinksIter);
        }
    }

    try {
        if (pthread_mutex_lock(&(gsCbData.mLock)) != 0) {
            cerr << "Cannot lock mutex in " << __FUNCTION__ << endl;
            return;
        }
        gsCbData.mpUrlScheduler->Feed(CheckedLinks);
        if (pthread_mutex_unlock(&(gsCbData.mLock)) != 0) {
            cerr << "Cannot unlock mutex in " << __FUNCTION__ << endl;
        }
    }
    catch (Naipax::Error &E) {
        cerr << E.What() << endl;
    }
}

void SigIntHandler(int Signal) {
    cerr << "SIGINT is received. Program exits." << endl;
    exit(0);
}

void OnAccept(int32_t Fd, void *pArg) {
    static const size_t BufSize = 4096;
    char Buf[BufSize];
    ssize_t ReadLen = 0;
    string ReadData;
    const string::size_type ReadDataMaxLen = 10485760; // 10 megabytes

    while (1) {
        ReadLen = read(Fd, Buf, BufSize);
        if (ReadLen > 0) {
            Buf[ReadLen] = '\0';
            if (ReadData.length() + ReadLen > ReadDataMaxLen) {
                break;
            }
            ReadData += string(Buf, ReadLen);
        }
        else if (ReadLen == 0) {
            break;
        }
    }

    stringstream ReadDataStream(ReadData);
    vector<string> Links;
    while (ReadDataStream.good() == true && ReadDataStream.eof() != true) {
        string UrlStr;
        ReadDataStream >> UrlStr;
        if (UrlStr.length() > 7 && string(UrlStr, 0, 7) == "http://") {
            cout << "URL: [" << UrlStr << "]" << endl;
            Url UrlObj(UrlStr);
            if (UrlObj.IsValid() != true) {
                continue;
            }
            UrlObj.Normalize();
            if (gsCbData.mpUrlHistory->Has(UrlObj.AsString()) != true) {
                Links.push_back(UrlObj.AsString());
            }
        }
    }

    if (Links.size() > 0) {
        if (pthread_mutex_lock(&(gsCbData.mLock)) != 0) {
            cerr << "Cannot lock mutex in " << __FUNCTION__ << endl;
            return;
        }
        gsCbData.mpUrlScheduler->Feed(Links);
        if (pthread_mutex_unlock(&(gsCbData.mLock)) != 0) {
            cerr << "Cannot unlock mutex in " << __FUNCTION__ << endl;
        }
    }

    if (gsStartTime == (time_t) -1) {
        gsStartTime = time(NULL);
    }
}

void* ServiceThreadCallback(void *pArg) {
    while (1) {
        try {
            gsCbData.mpServer->Process(NULL);
        }
        catch (Naipax::Error &E) {
            cerr << E.What() << endl;
        }
    }
    return NULL;
}

void* FetcherThreadCallback(void *pArg) {
    Connection *pConn = gsCbData.mpConn;
    PpqScheduler<UrlDispatcher> *pUrlScheduler = gsCbData.mpUrlScheduler;

    vector<string> UrlVector;
    while (1) {
        if (UrlVector.size() == 0) {
            try {
                if (pthread_mutex_lock(&(gsCbData.mLock)) != 0) {
                    cerr << "Cannot lock mutex in " << __FUNCTION__ << endl;
                    continue;
                }
                pUrlScheduler->Queue()->Dequeue(UrlVector, 100);
                if (pthread_mutex_unlock(&(gsCbData.mLock)) != 0) {
                    cerr << "Cannot unlock mutex in " << __FUNCTION__ << endl;
                }
            }
            catch (Naipax::Error &E) {
                cerr << E.What() << endl;
            }
        }
        else {
            try {
                string &rUrl = UrlVector.back();
                if (rUrl.length() > 0
                    && gsCbData.mpUrlHistory->Has(rUrl) != true) {
                    pConn->Add(rUrl, gsCbData.mpConf->mConnectionTimeout);
                    gsCbData.mpUrlHistory->Add(rUrl);
                    cout << rUrl << " is added to connection ..." << endl;
                }
                UrlVector.pop_back();
            }
            catch (Naipax::Error &E) {
                if (E.ErrorCode() != Error::BUFFER_IS_FULL) {
                    UrlVector.pop_back();
                    cerr << E.What() << endl;
                }
            }
        }

        try {
            pConn->OneNbLoop();
        }
        catch (Naipax::Error &E) {
            cerr << E.What() << endl;
        }
    }
    return NULL;
}

void StartNaipax() {
    pthread_t FetcherThread;
    pthread_t ServiceThread;

    if (pthread_mutex_init(&(gsCbData.mLock), NULL) != 0) {
        cerr << "Cannot initialize mutex" << endl;
        exit(-1);
    }
    
    if (pthread_create(&FetcherThread, NULL,
                       FetcherThreadCallback, NULL) != 0) {
        cerr << "Cannot create FetchThread" << endl;
        exit(-1);
    }
    if (pthread_create(&ServiceThread, NULL,
                       ServiceThreadCallback, NULL) != 0) {
        cerr << "Cannot create ServiceThread" << endl;
        exit(-1);
    }
    
    if (pthread_join(FetcherThread, NULL) != 0) {
        cerr << "Cannot join FetcherThread" << endl;
        exit(-1);
    }
    if (pthread_join(ServiceThread, NULL) != 0) {
        cerr << "Cannot join ServiceThread" << endl;
        exit(-1);
    }

    if (pthread_mutex_destroy(&(gsCbData.mLock)) != 0) {
        cerr << "Cannot destroy mutex" << endl;
        exit(-1);
    }
}

int32_t main(int32_t Argc, const char *pArgv[]) {
    cerr << "Naipax version: " << NAIPAX_VERSION << endl;
    if (Argc != 2) {
        cerr << "Usage: naipax config_file" << endl;
        exit(-1);
    }
    cerr << "Naipax is started" << endl << endl;

    struct sigaction SigAction;
    memset(&SigAction, 0, sizeof(SigAction));
    SigAction.sa_handler = SigIntHandler;
    
    if (sigaction(SIGINT, &SigAction, NULL) != 0) {
        cerr << strerror(errno) << endl;
        exit(-1);
    }
    
    try {
        Config Conf;
        Conf.ParseFile((const char*) pArgv[1]);

        ServiceServer Server;
        Server.mOnAccept = OnAccept;
        Server.Setup(Conf.mServicePort);
        
        Connection Conn((uint32_t) Conf.mCrawlerConnections);
        Conn.mpResponseHandler = ResponseHandler;
        Conn.UserAgent() = Conf.mCrawlerAgent;

        History UrlHist(Conf.mUrlHistoryFile.c_str());
        PpqScheduler<UrlDispatcher> UrlScheduler(Conf.mUrlPool.c_str(), 10);
        
        gsCbData.mpConf = &Conf;
        gsCbData.mpConn = &Conn;
        gsCbData.mpServer = &Server;
        gsCbData.mpUrlScheduler = &UrlScheduler;
        gsCbData.mpUrlHistory = &UrlHist;

        StartNaipax();
    }
    catch (Naipax::Error &E) {
        cerr << E.What() << endl;
        exit(-1);
    }
    
    return 0;
}
